/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.obs;

import com.facebook.airlift.log.Logger;
import com.facebook.presto.hive.filesystem.ExtendedFileSystem;
import com.facebook.presto.memory.context.AggregatedMemoryContext;
import com.facebook.presto.memory.context.LocalMemoryContext;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Splitter;
import com.google.common.collect.AbstractSequentialIterator;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterators;
import com.google.common.io.Closer;
import com.google.common.net.MediaType;
import com.obs.services.BasicObsCredentialsProvider;
import com.obs.services.DefaultCredentialsProviderChain;
import com.obs.services.IObsCredentialsProvider;
import com.obs.services.OBSCredentialsProviderChain;
import com.obs.services.ObsClient;
import com.obs.services.ObsConfiguration;
import com.obs.services.exception.ObsException;
import com.obs.services.internal.ServiceException;
import com.obs.services.model.AbortMultipartUploadRequest;
import com.obs.services.model.CompleteMultipartUploadRequest;
import com.obs.services.model.CopyObjectRequest;
import com.obs.services.model.GetObjectMetadataRequest;
import com.obs.services.model.GetObjectRequest;
import com.obs.services.model.InitiateMultipartUploadRequest;
import com.obs.services.model.InitiateMultipartUploadResult;
import com.obs.services.model.ListObjectsRequest;
import com.obs.services.model.ObjectListing;
import com.obs.services.model.ObjectMetadata;
import com.obs.services.model.ObsObject;
import com.obs.services.model.PartEtag;
import com.obs.services.model.PutObjectBasicRequest;
import com.obs.services.model.PutObjectRequest;
import com.obs.services.model.SseCHeader;
import com.obs.services.model.SseKmsHeader;
import com.obs.services.model.UploadPartRequest;
import com.obs.services.model.UploadPartResult;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.BufferedFSInputStream;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.URI;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;

import static com.facebook.airlift.concurrent.Threads.threadsNamed;
import static com.facebook.presto.hive.RetryDriver.retry;
import static com.facebook.presto.memory.context.AggregatedMemoryContext.newSimpleAggregatedMemoryContext;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_ACCESS_KEY;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_ACL_TYPE;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_CONNECT_TIMEOUT;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_ENDPOINT;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_BACKOFF_TIME;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_CLIENT_RETRIES;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_CONNECTIONS;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_ERROR_RETRIES;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_RETRY_TIME;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_PATH_STYLE_ACCESS;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SECRET_KEY;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SECURITY_TOKEN;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SKIP_GLACIER_OBJECTS;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SOCKET_TIMEOUT;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SSE_C_KEY_BASE64;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SSE_ENABLED;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SSE_KMS_KEY_ID;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SSE_TYPE;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_STAGING_DIRECTORY;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_STORAGE_CLASS;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_STREAMING_UPLOAD_ENABLED;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_STREAMING_UPLOAD_PART_SIZE;
import static com.google.common.base.MoreObjects.firstNonNull;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkPositionIndexes;
import static com.google.common.base.Strings.isNullOrEmpty;
import static com.google.common.base.Strings.nullToEmpty;
import static com.google.common.base.Throwables.throwIfInstanceOf;
import static com.google.common.base.Throwables.throwIfUnchecked;
import static com.google.common.collect.Iterables.toArray;
import static com.google.common.hash.Hashing.md5;
import static com.obs.services.model.StorageClassEnum.COLD;
import static com.obs.services.model.StorageClassEnum.DEEP_ARCHIVE;
import static io.airlift.units.DataSize.Unit.MEGABYTE;
import static java.lang.Math.max;
import static java.lang.Math.min;
import static java.lang.Math.toIntExact;
import static java.lang.String.format;
import static java.lang.System.arraycopy;
import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
import static java.net.HttpURLConnection.HTTP_FORBIDDEN;
import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
import static java.nio.file.Files.createDirectories;
import static java.nio.file.Files.createTempFile;
import static java.util.Objects.requireNonNull;
import static java.util.concurrent.Executors.newCachedThreadPool;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toList;
import static org.apache.hadoop.fs.FSExceptionMessages.CANNOT_SEEK_PAST_EOF;
import static org.apache.hadoop.fs.FSExceptionMessages.NEGATIVE_SEEK;
import static org.apache.hadoop.fs.FSExceptionMessages.STREAM_IS_CLOSED;

public class PrestoObsFileSystem
        extends ExtendedFileSystem
{
    private static final Logger log = Logger.get(PrestoObsFileSystem.class);

    private static final PrestoObsFileSystemStats STATS = new PrestoObsFileSystemStats();

    private static final String DIRECTORY_SUFFIX = "_$folder$";

    private static final DataSize BLOCK_SIZE = new DataSize(32, MEGABYTE);

    private static final DataSize MAX_SKIP_SIZE = new DataSize(1, MEGABYTE);

    private static final String PATH_SEPARATOR = "/";

    private static final Duration BACKOFF_MIN_SLEEP = new Duration(1, SECONDS);
    private static final int HTTP_RANGE_NOT_SATISFIABLE = 416;
    private static final MediaType DIRECTORY_MEDIA_TYPE = MediaType.create("application", "x-directory");

    private static final Set<String> GLACIER_STORAGE_CLASSES = ImmutableSet.of(COLD.toString(), DEEP_ARCHIVE.toString());

    static final String NO_SUCH_KEY_ERROR_CODE = "NoSuchKey";

    static final String NO_SUCH_BUCKET_ERROR_CODE = "NoSuchBucket";

    private URI uri;
    private Path workingDirectory;
    private ObsClient obsClient;
    private IObsCredentialsProvider credentialsProvider;
    private File stagingDirectory;
    private int maxAttempts;
    private Duration maxBackoffTime;
    private Duration maxRetryTime;
    private boolean sseEnabled;
    private PrestoObsSseType sseType;
    private String sseKmsKeyId;
    private String sseCKeyBase64;
    private PrestoObsAclType obsAclType;
    private boolean skipGlacierObjects;
    private boolean streamingUploadEnabled;
    private int streamingUploadPartSize;
    private PrestoObsStorageClass obsStorageClass;

    private final ExecutorService uploadExecutor = newCachedThreadPool(threadsNamed("obs-upload-%s"));

    @Override
    public void initialize(URI uri, Configuration conf)
            throws IOException
    {
        requireNonNull(uri, "uri is null");
        requireNonNull(conf, "conf is null");
        super.initialize(uri, conf);
        setConf(conf);

        this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
        this.workingDirectory = new Path(PATH_SEPARATOR).makeQualified(this.uri, new Path(PATH_SEPARATOR));

        HiveObsConfig defaults = new HiveObsConfig();
        this.stagingDirectory = new File(conf.get(OBS_STAGING_DIRECTORY, defaults.getObsStagingDirectory().toString()));
        this.maxAttempts = conf.getInt(OBS_MAX_CLIENT_RETRIES, defaults.getObsMaxClientRetries()) + 1;
        this.maxBackoffTime = Duration.valueOf(conf.get(OBS_MAX_BACKOFF_TIME, defaults.getObsMaxBackoffTime().toString()));
        this.maxRetryTime = Duration.valueOf(conf.get(OBS_MAX_RETRY_TIME, defaults.getObsMaxRetryTime().toString()));
        this.sseEnabled = conf.getBoolean(OBS_SSE_ENABLED, defaults.isObsSseEnabled());
        this.sseType = PrestoObsSseType.valueOf(conf.get(OBS_SSE_TYPE, defaults.getObsSseType().name()));
        this.sseKmsKeyId = conf.get(OBS_SSE_KMS_KEY_ID, defaults.getObsSseKmsKeyId());
        this.sseCKeyBase64 = conf.get(OBS_SSE_C_KEY_BASE64, defaults.getObsSseCKeyBase64());
        this.obsAclType = PrestoObsAclType.valueOf(conf.get(OBS_ACL_TYPE, defaults.getObsAclType().name()));
        this.skipGlacierObjects = conf.getBoolean(OBS_SKIP_GLACIER_OBJECTS, defaults.isSkipGlacierObjects());
        this.streamingUploadEnabled = conf.getBoolean(OBS_STREAMING_UPLOAD_ENABLED, defaults.isObsStreamingUploadEnabled());
        this.streamingUploadPartSize = toIntExact(conf.getLong(OBS_STREAMING_UPLOAD_PART_SIZE, defaults.getObsStreamingPartSize().toBytes()));
        this.obsStorageClass = conf.getEnum(OBS_STORAGE_CLASS, defaults.getObsStorageClass());

        Duration connectTimeout = Duration.valueOf(conf.get(OBS_CONNECT_TIMEOUT, defaults.getObsConnectTimeout().toString()));
        Duration socketTimeout = Duration.valueOf(conf.get(OBS_SOCKET_TIMEOUT, defaults.getObsSocketTimeout().toString()));
        int maxConnections = conf.getInt(OBS_MAX_CONNECTIONS, defaults.getObsMaxConnections());
        int maxErrorRetries = conf.getInt(OBS_MAX_ERROR_RETRIES, defaults.getObsMaxErrorRetries());
        boolean isPathStyleAccess = conf.getBoolean(OBS_PATH_STYLE_ACCESS, defaults.isObsPathStyleAccess());

        ObsConfiguration obsConfiguration = new ObsConfiguration();
        obsConfiguration.setConnectionTimeout((int) connectTimeout.toMillis());
        obsConfiguration.setSocketTimeout((int) socketTimeout.toMillis());
        obsConfiguration.setMaxConnections(maxConnections);
        obsConfiguration.setMaxErrorRetry(maxErrorRetries);
        obsConfiguration.setPathStyle(isPathStyleAccess);

        this.credentialsProvider = createObsCredentialsProvider(uri,conf);
        this.obsClient = createsObsClient(conf,obsConfiguration);
    }

    @Override
    public void close()
            throws IOException
    {
        try (Closer closer = Closer.create()) {
            closer.register(super::close);
            closer.register(obsClient::close);
        }
    }

    @Override
    public URI getUri()
    {
        return uri;
    }

    @Override
    public String getScheme()
    {
        return uri.getScheme();
    }

    @Override
    public Path getWorkingDirectory() {
        return workingDirectory;
    }

    @Override
    public void setWorkingDirectory(Path path) {
        this.workingDirectory = path;
    }

    @Override
    public FileStatus[] listStatus(Path path)
            throws IOException
    {
        STATS.newListStatusCall();
        List<LocatedFileStatus> list = new ArrayList<>();
        RemoteIterator<LocatedFileStatus> iterator = listLocatedStatus(path);
        while (iterator.hasNext()) {
            list.add(iterator.next());
        }
        return toArray(list, LocatedFileStatus.class);
    }

    @Override
    public RemoteIterator<LocatedFileStatus> listFiles(Path path, boolean recursive)
    {
        // Either a single level or full listing, depending on the recursive flag, no "directories" are included
        return new ObsObjectsRemoteIterator(listPrefix(path, OptionalInt.empty(), recursive ? ListingMode.RECURSIVE_FILES_ONLY : ListingMode.SHALLOW_FILES_ONLY));
    }

    @Override
    public RemoteIterator<LocatedFileStatus> listLocatedStatus(Path path)
    {
        STATS.newListLocatedStatusCall();
        return new ObsObjectsRemoteIterator(listPrefix(path, OptionalInt.empty(), ListingMode.SHALLOW_ALL));
    }

    private static final class ObsObjectsRemoteIterator
            implements RemoteIterator<LocatedFileStatus>
    {

        private final Iterator<LocatedFileStatus> iterator;

        public ObsObjectsRemoteIterator(Iterator<LocatedFileStatus> iterator)
        {
            this.iterator = requireNonNull(iterator, "iterator is null");
        }

        @Override
        public boolean hasNext()
                throws IOException
        {
            try {
                return iterator.hasNext();
            }
            catch (ObsException e) {
                throw new IOException(e);
            }
        }

        @Override
        public LocatedFileStatus next()
                throws IOException
        {
            try {
                return iterator.next();
            }
            catch (ObsException e) {
                throw new IOException(e);
            }
        }
    }

    @Override
    public FileStatus getFileStatus(Path path)
            throws IOException
    {
        if (path.getName().isEmpty()) {
            // the bucket root requires special handling
            if (getObsObjectMetadata(path) != null) {
                return new FileStatus(0, true, 1, 0, 0, qualifiedPath(path));
            }
            throw new FileNotFoundException("File does not exist: " + path);
        }
        ObjectMetadata metadata = getObsObjectMetadata(path);
        if (metadata == null) {
            // check if this path is a directory
            Iterator<LocatedFileStatus> iterator = listPrefix(path, OptionalInt.of(1), ListingMode.SHALLOW_ALL);
            if (iterator.hasNext()) {
                return new FileStatus(0, true, 1, 0, 0, qualifiedPath(path));
            }
            throw new FileNotFoundException("File does not exist: " + path);
        }
        return new FileStatus(
                getObjectSize(path, metadata),
                MediaType.parse(metadata.getContentType()).is(DIRECTORY_MEDIA_TYPE),
                1,
                BLOCK_SIZE.toBytes(),
                lastModifiedTime(metadata),
                qualifiedPath(path));
    }

    private static long getObjectSize(Path path, ObjectMetadata metadata)
            throws IOException
    {
        return metadata.getContentLength() == null ? 0L : metadata.getContentLength();
    }

    @Override
    public FSDataInputStream open(Path path, int bufferSize) throws IOException
    {
        return new FSDataInputStream(
                new BufferedFSInputStream(
                        new PrestoObsInputStream(obsClient, getBucketName(uri), path, maxAttempts, maxBackoffTime, maxRetryTime, sseEnabled, sseType, sseCKeyBase64),
                        bufferSize));
    }

    @Override
    public FSDataOutputStream create(Path path, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress)
            throws IOException
    {
        if ((!overwrite) && exists(path)) {
            throw new IOException("File already exists:" + path);
        }
        String bucketName = getBucketName(uri);
        String key = keyFromPath(qualifiedPath(path));

        if (streamingUploadEnabled) {
            Supplier<String> uploadIdFactory =  () -> initMultipartUpload(bucketName, key).getUploadId();

            return new FSDataOutputStream(
                new PrestoObsStreamingOutputStream(obsClient,
                        bucketName,
                        key,
                        sseEnabled,
                        sseType,
                        sseKmsKeyId,
                        sseCKeyBase64,
                        obsAclType,
                        obsStorageClass,
                        uploadIdFactory,
                        uploadExecutor,
                        streamingUploadPartSize,
                        newSimpleAggregatedMemoryContext()
                        ),
                statistics);
        }

        if (!stagingDirectory.exists()) {
            createDirectories(stagingDirectory.toPath());
        }
        if (!stagingDirectory.isDirectory()) {
            throw new IOException("Configured staging path is not a directory: " + stagingDirectory);
        }
        File tempFile = createTempFile(stagingDirectory.toPath(), "presto-obs-", ".tmp").toFile();
        return new FSDataOutputStream(
                new PrestoObsOutputStream(obsClient,
                        bucketName,
                        key,
                        tempFile,
                        sseEnabled,
                        sseType,
                        sseKmsKeyId,
                        sseCKeyBase64,
                        obsAclType,
                        obsStorageClass),
                statistics);
    }


    @Override
    public FSDataOutputStream append(Path path, int i, Progressable progressable) throws IOException
    {
        throw new UnsupportedOperationException("append");
    }

    @Override
    public boolean rename(Path src, Path dst) throws IOException
    {
        boolean srcDirectory;
        try {
            srcDirectory = directory(src);
        }
        catch (FileNotFoundException e) {
            return false;
        }

        try {
            if (!directory(dst)) {
                // cannot copy a file to an existing file
                return false;
            }
            // move source under destination directory
            dst = new Path(dst, src.getName());
        }
        catch (FileNotFoundException e) {
            // destination does not exist
        }

        if (keysEqual(src, dst)) {
            return false;
        }

        if (srcDirectory) {
            for (FileStatus file : listStatus(src)) {
                rename(file.getPath(), new Path(dst, file.getPath().getName()));
            }
            deleteObject(keyFromPath(src) + DIRECTORY_SUFFIX);
        }
        else {
            CopyObjectRequest request = new CopyObjectRequest(getBucketName(uri), keyFromPath(src), getBucketName(uri), keyFromPath(dst));
            if (sseEnabled) {
                switch (sseType) {
                    case KMS:
                        SseKmsHeader sseKmsHeader = new SseKmsHeader();
                        if (sseKmsKeyId != null) {
                            sseKmsHeader.setKmsKeyId(sseKmsKeyId);
                        }
                        request.setSseKmsHeader(sseKmsHeader);
                        break;
                    case C:
                        if (sseCKeyBase64 != null) {
                            SseCHeader sseCHeader = new SseCHeader();
                            sseCHeader.setSseCKeyBase64(sseCKeyBase64);
                            request.setSseCHeader(sseCHeader);
                            request.setSseCHeaderSource(sseCHeader);
                        }
                        break;
                }
            }
            obsClient.copyObject(request);
            delete(src, true);
        }
        return true;
    }


    @Override
    public boolean delete(Path path, boolean recursive) throws IOException
    {
        try {
            if (!directory(path)) {
                return deleteObject(keyFromPath(path));
            }
        }
        catch (FileNotFoundException e) {
            return false;
        }

        if (!recursive) {
            throw new IOException("Directory " + path + " is not empty");
        }

        for (FileStatus file : listStatus(path)) {
            delete(file.getPath(), true);
        }
        deleteObject(keyFromPath(path) + DIRECTORY_SUFFIX);

        return true;
    }


    private boolean directory(Path path)
            throws IOException
    {
        return getFileStatus(path).isDirectory();
    }

    private boolean deleteObject(String key)
    {
        try {
            obsClient.deleteObject(getBucketName(uri), key);
            return true;
        }
        catch (ObsException e) {
            return false;
        }
    }

    @Override
    public boolean mkdirs(Path path, FsPermission fsPermission) throws IOException
    {
        // no need to do anything for obs
        return true;
    }

    private enum ListingMode {
        SHALLOW_ALL, // Shallow listing of files AND directories
        SHALLOW_FILES_ONLY, //仅当前目录的文件
        RECURSIVE_FILES_ONLY;  //递归文件

        public boolean isFilesOnly() {
            return (this == SHALLOW_FILES_ONLY || this == RECURSIVE_FILES_ONLY);
        }
    }


    private Iterator<LocatedFileStatus> listPrefix(Path path, OptionalInt initialMaxKeys, ListingMode mode)
    {
        String key = keyFromPath(path);
        if (!key.isEmpty()) {
            key += PATH_SEPARATOR;
        }

        ListObjectsRequest request = new ListObjectsRequest();
        request.setBucketName(getBucketName(uri));
        request.setPrefix(key);
        request.setDelimiter(mode == ListingMode.RECURSIVE_FILES_ONLY ? null : PATH_SEPARATOR);

        if(initialMaxKeys.isPresent()) {
            request.setMaxKeys(initialMaxKeys.getAsInt());
        }

        STATS.newListObjectsCall();
        Iterator<ObjectListing> listings = new AbstractSequentialIterator<ObjectListing>(obsClient.listObjects(request))
        {
            @Override
            protected ObjectListing computeNext(ObjectListing previous)
            {
                if (!previous.isTruncated()) {
                    return null;
                }
                // Clear any max keys set for the initial request before submitting subsequent requests. Values < 0
                // are not sent in the request and the default limit is used
                request.setMaxKeys(-1);
                request.setMarker(previous.getNextMarker());

                return obsClient.listObjects(request);
            }
        };

        Iterator<LocatedFileStatus> result = Iterators.concat(Iterators.transform(listings, this::statusFromListing));
        if (mode.isFilesOnly()) {
            //  Even recursive listing can still contain empty "directory" objects, must filter them out
            result = Iterators.filter(result, LocatedFileStatus::isFile);
        }
        return result;
    }


    private Iterator<LocatedFileStatus> statusFromListing(ObjectListing listing)
    {
        List<String> prefixes = listing.getCommonPrefixes();
        List<ObsObject> objects = listing.getObjects();
        if (prefixes.isEmpty()) {
            return statusFromObjects(objects);
        }
        if (objects.isEmpty()) {
            return statusFromPrefixes(prefixes);
        }
        return Iterators.concat(
                statusFromPrefixes(prefixes),
                statusFromObjects(objects));
    }


    private Iterator<LocatedFileStatus> statusFromPrefixes(List<String> prefixes)
    {
        List<LocatedFileStatus> list = new ArrayList<LocatedFileStatus>(prefixes.size());
        for (String prefix : prefixes) {
            Path path = qualifiedPath(new Path(PATH_SEPARATOR + prefix));
            FileStatus status = new FileStatus(0, true, 1, 0, 0, path);
            list.add(createLocatedFileStatus(status));
        }
        return list.iterator();
    }

    private Iterator<LocatedFileStatus> statusFromObjects(List<ObsObject> objects)
    {
        // NOTE: for encrypted objects, ObsObject.size() used below is NOT correct,
        // however, to get the correct size we'd need to make an additional request to get
        // user metadata, and in this case it doesn't matter.
        return objects.stream()
                .filter(object -> !object.getObjectKey().endsWith(PATH_SEPARATOR))
                .filter(object -> !skipGlacierObjects || !isGlacierObject(object))
                .filter(object -> !isHadoopFolderMarker(object))
                .map(object -> new FileStatus(
                        object.getMetadata().getContentLength() == null ? 0 : object.getMetadata().getContentLength(),
                        false,
                        1,
                        BLOCK_SIZE.toBytes(),
                        object.getMetadata().getLastModified().getTime(),
                        qualifiedPath(new Path(PATH_SEPARATOR + object.getObjectKey()))))
                .map(this::createLocatedFileStatus)
                .iterator();
    }

    private boolean isGlacierObject(ObsObject object)
    {
        return GLACIER_STORAGE_CLASSES.contains(object.getMetadata().getObjectStorageClass().toString());
    }

    private boolean isHadoopFolderMarker(ObsObject object)
    {
        return (object.getObjectKey().endsWith(DIRECTORY_SUFFIX) && (object.getMetadata().getContentLength() == null || object.getMetadata().getContentLength().longValue() == 0));
    }

    /**
     * This exception is for stopping retries for OBS calls that shouldn't be retried.
     */
    @VisibleForTesting
    static class UnrecoverableObsOperationException
            extends IOException
    {
        public UnrecoverableObsOperationException(Path path, Throwable cause)
        {
            // append the path info to the message
            super(format("%s (Path: %s)", cause, path), cause);
        }

        public UnrecoverableObsOperationException(String bucket, String key, Throwable cause)
        {
            // append bucket and key to the message
            super(format("%s (Bucket: %s, Key: %s)", cause, bucket, key));
        }
    }

    @VisibleForTesting
    ObjectMetadata getObsObjectMetadata(Path path)
            throws IOException
    {
        String bucketName = getBucketName(uri);
        String key = keyFromPath(path);
        ObjectMetadata obsObjectMetadata = getObsObjectMetadata(path, bucketName, key);
        if (obsObjectMetadata == null && !key.isEmpty()) {
            obsObjectMetadata =  getObsObjectMetadata(path, bucketName, key + PATH_SEPARATOR);
            if(obsObjectMetadata != null){
                obsObjectMetadata.setContentType(DIRECTORY_MEDIA_TYPE.toString());
                return obsObjectMetadata;
            }
        }
        return obsObjectMetadata;
    }


    private ObjectMetadata getObsObjectMetadata(Path path, String bucketName, String key)
            throws IOException
    {
        try {
            return retry()
                    .maxAttempts(maxAttempts)
                    .exponentialBackoff(BACKOFF_MIN_SLEEP, maxBackoffTime, maxRetryTime, 2.0)
                    .stopOn(InterruptedException.class, UnrecoverableObsOperationException.class, ObsException.class)
                    .onRetry(STATS::newGetMetadataRetry)
                    .run("getObsObjectMetadata", () -> {
                        try {
                            STATS.newMetadataCall();
                            GetObjectMetadataRequest request = new GetObjectMetadataRequest(bucketName, key);

                            if (sseEnabled) {
                                switch (sseType) {
                                    case C:
                                        if (sseCKeyBase64 != null) {
                                            SseCHeader sseCHeader = new SseCHeader();
                                            sseCHeader.setSseCKeyBase64(sseCKeyBase64);
                                            request.setSseCHeader(sseCHeader);
                                        }
                                        break;
                                }
                            }
                            return obsClient.getObjectMetadata(request);
                        }
                        catch (RuntimeException e) {
                            STATS.newGetMetadataError();
                            if (e instanceof ObsException) {
                                ObsException obsException = (ObsException) e;
                                switch (obsException.getResponseCode()) {
                                    case HTTP_NOT_FOUND:
                                        return null;
                                    case HTTP_FORBIDDEN:
                                    case HTTP_BAD_REQUEST:
                                        throw new UnrecoverableObsOperationException(path, e);
                                }
                            }
                            if (e instanceof ServiceException) {
                                ServiceException serviceException = (ServiceException) e;
                                int response = serviceException.getResponseCode();
                                switch (response) {
                                    case HTTP_NOT_FOUND:
                                        return null;
                                    case HTTP_FORBIDDEN:
                                    case HTTP_BAD_REQUEST:
                                        throw new UnrecoverableObsOperationException(path, e);
                                }
                            }
                            throw e;
                        }
                    });
        }
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
        catch (Exception e) {
            throwIfInstanceOf(e, IOException.class);
            throwIfUnchecked(e);
            throw new RuntimeException(e);
        }
    }

    private Path qualifiedPath(Path path)
    {
        return path.makeQualified(this.uri, getWorkingDirectory());
    }


    private LocatedFileStatus createLocatedFileStatus(FileStatus status)
    {
        try {
            BlockLocation[] fakeLocation = getFileBlockLocations(status, 0, status.getLen());
            return new LocatedFileStatus(status, fakeLocation);
        }
        catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    private static long lastModifiedTime(ObjectMetadata metadata)
    {
        Date date = metadata.getLastModified();
        return (date != null) ? date.getTime() : 0;
    }

    private static boolean keysEqual(Path p1, Path p2)
    {
        return keyFromPath(p1).equals(keyFromPath(p2));
    }

    public static String keyFromPath(Path path)
    {
        checkArgument(path.isAbsolute(), "Path is not absolute: %s", path);
        String key = nullToEmpty(path.toUri().getPath());
        if (key.startsWith(PATH_SEPARATOR)) {
            key = key.substring(PATH_SEPARATOR.length());
        }
        if (key.endsWith(PATH_SEPARATOR)) {
            key = key.substring(0, key.length() - PATH_SEPARATOR.length());
        }
        return key;
    }

    private ObsClient createsObsClient(Configuration hadoopConfig, ObsConfiguration obsConfiguration)
    {
        String endpoint = hadoopConfig.get(OBS_ENDPOINT);
        obsConfiguration.setEndPoint(endpoint);
        return new ObsClient(credentialsProvider,obsConfiguration);
    }

    private IObsCredentialsProvider createObsCredentialsProvider(URI uri, Configuration conf)
    {
        Optional<IObsCredentialsProvider> credentials = getEmbeddedObsCredentials(uri);
        if (credentials.isPresent()) {
            return credentials.get();
        }
        // use configured credentials or default chain with optional role
        IObsCredentialsProvider provider = getObsCredentials(conf)
                .map(value -> (IObsCredentialsProvider) new DefaultCredentialsProviderChain(value))
                .orElse(new OBSCredentialsProviderChain());
        return provider;
    }


    private static Optional<IObsCredentialsProvider> getEmbeddedObsCredentials(URI uri)
    {
        String userInfo = nullToEmpty(uri.getUserInfo());
        List<String> parts = Splitter.on(':').limit(2).splitToList(userInfo);
        if (parts.size() == 2) {
            String accessKey = parts.get(0);
            String secretKey = parts.get(1);
            if (!accessKey.isEmpty() && !secretKey.isEmpty()) {
                return Optional.of(new BasicObsCredentialsProvider(accessKey, secretKey));
            }
        }
        return Optional.empty();
    }

    private static Optional<IObsCredentialsProvider> getObsCredentials(Configuration conf)
    {
        String accessKey = conf.get(OBS_ACCESS_KEY);
        String secretKey = conf.get(OBS_SECRET_KEY);

        if (isNullOrEmpty(accessKey) || isNullOrEmpty(secretKey)) {
            return Optional.empty();
        }

        String securityToken = conf.get(OBS_SECURITY_TOKEN);
        if (!isNullOrEmpty(securityToken)) {
            return Optional.of(new BasicObsCredentialsProvider(accessKey, secretKey, securityToken));
        }
        return Optional.of(new BasicObsCredentialsProvider(accessKey, secretKey));
    }

    private static class PrestoObsInputStream
            extends FSInputStream
    {
        private final ObsClient obsClient;
        private final String host;
        private final Path path;
        private final int maxAttempts;
        private final Duration maxBackoffTime;
        private final Duration maxRetryTime;

        private final AtomicBoolean closed = new AtomicBoolean();

        private InputStream in;
        private long streamPosition;
        private long nextReadPosition;
        private boolean sseEnabled;
        private PrestoObsSseType sseType;

        private String sseCKeyBase64;

        public PrestoObsInputStream(ObsClient obsClient, String host, Path path, int maxAttempts, Duration maxBackoffTime,
                                    Duration maxRetryTime, boolean sseEnabled, PrestoObsSseType sseType, String sseCKeyBase64)
        {
            this.obsClient = requireNonNull(obsClient, "obsClient is null");
            this.host = requireNonNull(host, "host is null");
            this.path = requireNonNull(path, "path is null");

            checkArgument(maxAttempts >= 0, "maxAttempts cannot be negative");
            this.maxAttempts = maxAttempts;
            this.maxBackoffTime = requireNonNull(maxBackoffTime, "maxBackoffTime is null");
            this.maxRetryTime = requireNonNull(maxRetryTime, "maxRetryTime is null");
            this.sseEnabled = sseEnabled;
            this.sseType = sseType;
            this.sseCKeyBase64 = sseCKeyBase64;
        }

        @Override
        public void close()
        {
            closed.set(true);
            closeStream();
        }

        @Override
        public int read(long position, byte[] buffer, int offset, int length)
                throws IOException
        {
            checkClosed();
            if (position < 0) {
                throw new EOFException(NEGATIVE_SEEK);
            }
            checkPositionIndexes(offset, offset + length, buffer.length);
            if (length == 0) {
                return 0;
            }

            try {
                return retry()
                        .maxAttempts(maxAttempts)
                        .exponentialBackoff(BACKOFF_MIN_SLEEP, maxBackoffTime, maxRetryTime, 2.0)
                        .stopOn(InterruptedException.class, UnrecoverableObsOperationException.class, EOFException.class, FileNotFoundException.class, ObsException.class)
                        .onRetry(STATS::newGetObjectRetry)
                        .run("getObsObject", () -> {
                            InputStream stream;
                            try {
                                GetObjectRequest request = new GetObjectRequest(host, keyFromPath(path));
                                request.setRangeStart(position);
                                request.setRangeEnd((position + length) - 1);
                                if (sseEnabled) {
                                    switch (sseType) {
                                        case C:
                                            if (sseCKeyBase64 != null) {
                                                SseCHeader sseCHeader = new SseCHeader();
                                                sseCHeader.setSseCKeyBase64(sseCKeyBase64);
                                                request.setSseCHeader(sseCHeader);
                                            }
                                            break;
                                    }
                                }

                                stream = obsClient.getObject(request).getObjectContent();
                            }
                            catch (RuntimeException e) {
                                STATS.newGetObjectError();
                                if (e instanceof ObsException) {
                                    switch (((ObsException) e).getResponseCode()) {
                                        case HTTP_RANGE_NOT_SATISFIABLE:
                                            throw new EOFException(CANNOT_SEEK_PAST_EOF);
                                        case HTTP_NOT_FOUND:
                                            throw new FileNotFoundException("File does not exist: " + path);
                                        case HTTP_FORBIDDEN:
                                        case HTTP_BAD_REQUEST:
                                            throw new UnrecoverableObsOperationException(path, e);
                                    }
                                }
                                if (e instanceof ServiceException) {
                                    switch (((ServiceException) e).getResponseCode()) {
                                        case HTTP_RANGE_NOT_SATISFIABLE:
                                            throw new EOFException(CANNOT_SEEK_PAST_EOF);
                                        case HTTP_NOT_FOUND:
                                            throw new FileNotFoundException("File does not exist: " + path);
                                        case HTTP_FORBIDDEN:
                                        case HTTP_BAD_REQUEST:
                                            throw new UnrecoverableObsOperationException(path, e);
                                    }
                                }
                                throw e;
                            }

                            STATS.connectionOpened();
                            try {
                                int read = 0;
                                while (read < length) {
                                    int n = stream.read(buffer, offset + read, length - read);
                                    if (n <= 0) {
                                        if (read > 0) {
                                            return read;
                                        }
                                        return -1;
                                    }
                                    read += n;
                                }
                                return read;
                            }
                            catch (Throwable t) {
                                STATS.newReadError(t);
                                abortStream(stream);
                                throw t;
                            }
                            finally {
                                STATS.connectionReleased();
                                stream.close();
                            }
                        });
            }
            catch (Exception e) {
                throw propagate(e);
            }
        }

        @Override
        public void seek(long pos)
                throws IOException
        {
            checkClosed();
            if (pos < 0) {
                throw new EOFException(NEGATIVE_SEEK);
            }

            // this allows a seek beyond the end of the stream but the next read will fail
            nextReadPosition = pos;
        }

        @Override
        public long getPos()
        {
            return nextReadPosition;
        }

        @Override
        public int read()
        {
            // This stream is wrapped with BufferedInputStream, so this method should never be called
            throw new UnsupportedOperationException();
        }

        @Override
        public int read(byte[] buffer, int offset, int length)
                throws IOException
        {
            checkClosed();
            try {
                int bytesRead = retry()
                        .maxAttempts(maxAttempts)
                        .exponentialBackoff(BACKOFF_MIN_SLEEP, maxBackoffTime, maxRetryTime, 2.0)
                        .stopOn(InterruptedException.class, UnrecoverableObsOperationException.class, ObsException.class, FileNotFoundException.class)
                        .onRetry(STATS::newReadRetry)
                        .run("readStream", () -> {
                            seekStream();
                            try {
                                return in.read(buffer, offset, length);
                            }
                            catch (Exception e) {
                                STATS.newReadError(e);
                                closeStream();
                                throw e;
                            }
                        });

                if (bytesRead != -1) {
                    streamPosition += bytesRead;
                    nextReadPosition += bytesRead;
                }
                return bytesRead;
            }
            catch (Exception e) {
                throw propagate(e);
            }
        }

        @Override
        public boolean seekToNewSource(long targetPos)
        {
            return false;
        }

        private void seekStream()
                throws IOException
        {
            if ((in != null) && (nextReadPosition == streamPosition)) {
                // already at specified position
                return;
            }

            if ((in != null) && (nextReadPosition > streamPosition)) {
                // seeking forwards
                long skip = nextReadPosition - streamPosition;
                if (skip <= max(in.available(), MAX_SKIP_SIZE.toBytes())) {
                    // already buffered or seek is small enough
                    try {
                        if (in.skip(skip) == skip) {
                            streamPosition = nextReadPosition;
                            return;
                        }
                    }
                    catch (IOException ignored) {
                        // will retry by re-opening the stream
                    }
                }
            }

            // close the stream and open at desired position
            streamPosition = nextReadPosition;
            closeStream();
            openStream();
        }

        private void openStream()
                throws IOException
        {
            if (in == null) {
                in = openStream(path, nextReadPosition);
                streamPosition = nextReadPosition;
                STATS.connectionOpened();
            }
        }

        private InputStream openStream(Path path, long start)
                throws IOException
        {
            try {
                return retry()
                        .maxAttempts(maxAttempts)
                        .exponentialBackoff(BACKOFF_MIN_SLEEP, maxBackoffTime, maxRetryTime, 2.0)
                        .stopOn(InterruptedException.class, UnrecoverableObsOperationException.class, FileNotFoundException.class, ObsException.class)
                        .onRetry(STATS::newGetObjectRetry)
                        .run("getObsObject", () -> {
                            try {
                                GetObjectRequest request = new GetObjectRequest(host, keyFromPath(path));
                                request.setRangeStart(start);

                                if (sseEnabled) {
                                    switch (sseType) {
                                        case C:
                                            if (sseCKeyBase64 != null) {
                                                SseCHeader sseCHeader = new SseCHeader();
                                                sseCHeader.setSseCKeyBase64(sseCKeyBase64);
                                                request.setSseCHeader(sseCHeader);
                                            }
                                            break;
                                    }
                                }
                                return obsClient.getObject(request).getObjectContent();
                            }
                            catch (RuntimeException e) {
                                STATS.newGetObjectError();
                                if (e instanceof ObsException) {
                                    switch (((ObsException) e).getResponseCode()) {
                                        case HTTP_RANGE_NOT_SATISFIABLE:
                                            // ignore request for start past end of object
                                            return new ByteArrayInputStream(new byte[0]);
                                        case HTTP_NOT_FOUND:
                                            throw new FileNotFoundException("File does not exist: " + path);
                                        case HTTP_FORBIDDEN:
                                        case HTTP_BAD_REQUEST:
                                            throw new UnrecoverableObsOperationException(path, e);
                                    }
                                }
                                if (e instanceof ServiceException) {
                                    switch (((ServiceException) e).getResponseCode()) {
                                        case HTTP_RANGE_NOT_SATISFIABLE:
                                            // ignore request for start past end of object
                                            return new ByteArrayInputStream(new byte[0]);
                                        case HTTP_NOT_FOUND:
                                            throw new FileNotFoundException("File does not exist: " + path);
                                        case HTTP_FORBIDDEN:
                                        case HTTP_BAD_REQUEST:
                                            throw new UnrecoverableObsOperationException(path, e);
                                    }
                                }
                                throw e;
                            }
                        });
            }
            catch (Exception e) {
                throw propagate(e);
            }
        }

        private void closeStream()
        {
            if (in != null) {
                abortStream(in);
                in = null;
                STATS.connectionReleased();
            }
        }

        private void checkClosed()
                throws IOException
        {
            if (closed.get()) {
                throw new IOException(STREAM_IS_CLOSED);
            }
        }

        private static void abortStream(InputStream in)
        {
            try {
                in.close();
            }
            catch (IOException | ObsException ignored) {
                // thrown if the current thread is in the interrupted state
            }
        }

        private static RuntimeException propagate(Exception e)
                throws IOException
        {
            if (e instanceof InterruptedException) {
                Thread.currentThread().interrupt();
                throw new InterruptedIOException();
            }
            throwIfInstanceOf(e, IOException.class);
            throwIfUnchecked(e);
            throw new IOException(e);
        }
    }

    private static class PrestoObsStreamingOutputStream
        extends OutputStream
    {
        private final ObsClient obsClient;
        private final String bucketName;
        private final String key;
        private final Supplier<String> uploadIdFactory;
        private final ExecutorService uploadExecutor;

        private final boolean sseEnabled;
        private final PrestoObsSseType sseType;
        private final String sseKmsKeyId;
        private final String sseCKeyBase64;
        private final PrestoObsAclType obsAclType;
        private final PrestoObsStorageClass obsStorageClass;

        private int currentPartNumber;

        private byte[] buffer;

        private int bufferSize;

        private boolean closed;

        private boolean failed;

        // Mutated and read by main thread; mutated just before scheduling upload to background thread (access does not need to be thread safe)
        private boolean multipartUploadStarted;
        // Mutated by background thread which does the multipart upload; read by both main thread and background thread;
        // Visibility ensured by memory barrier via inProgressUploadFuture
        private Optional<String> uploadId = Optional.empty();
        private Future<UploadPartResult> inProgressUploadFuture;
        private final List<UploadPartResult> parts = new ArrayList<>();
        private final int partSize;
        private int initialBufferSize;
        private final LocalMemoryContext memoryContext;

        public PrestoObsStreamingOutputStream(
                ObsClient obsClient,
                String bucketName,
                String key,
                boolean sseEnabled,
                PrestoObsSseType sseType,
                String sseKmsKeyId,
                String sseCKeyBase64,
                PrestoObsAclType obsAclType,
                PrestoObsStorageClass obsStorageClass,
                Supplier<String> uploadIdFactory,
                ExecutorService uploadExecutor,
                int partSize,
                AggregatedMemoryContext memoryContext)
        {
            STATS.uploadStarted();

            this.obsClient = requireNonNull(obsClient, "obsClient is null");
            this.bucketName = requireNonNull(bucketName, "bucketName is null");
            this.key = requireNonNull(key, "key is null");
            this.partSize = partSize;

            this.sseEnabled = sseEnabled;
            this.sseType = requireNonNull(sseType, "sseType is null");
            this.sseKmsKeyId = sseKmsKeyId;
            this.sseCKeyBase64 = sseCKeyBase64;
            this.obsAclType = obsAclType;
            this.obsStorageClass = obsStorageClass;

            this.uploadIdFactory = requireNonNull(uploadIdFactory, "uploadIdFactory is null");
            this.uploadExecutor = requireNonNull(uploadExecutor, "uploadExecutor is null");
            this.buffer = new byte[0];
            this.initialBufferSize = 64;

            this.memoryContext = requireNonNull(memoryContext, "memoryContext is null")
                    .newLocalMemoryContext(PrestoObsStreamingOutputStream.class.getSimpleName());
        }

        @Override
        public void write(int b)
                throws IOException
        {
            ensureExtraBytesCapacity(1);
            flushBuffer(false);
            buffer[bufferSize] = (byte) b;
            bufferSize++;
        }

        @Override
        public void write(byte[] bytes, int offset, int length)
                throws IOException
        {
            while (length > 0) {
                ensureExtraBytesCapacity(min(partSize - bufferSize, length));
                int copied = min(buffer.length - bufferSize, length);
                arraycopy(bytes, offset, buffer, bufferSize, copied);
                bufferSize += copied;

                flushBuffer(false);

                offset += copied;
                length -= copied;
            }
        }

        @Override
        public void flush()
                throws IOException
        {
            flushBuffer(false);
        }

        @Override
        public void close()
                throws IOException
        {
            if (closed) {
                return;
            }
            closed = true;

            if (failed) {
                try {
                    abortUpload();
                    return;
                }
                catch (RuntimeException e) {
                    throw new IOException(e);
                }
            }

            try {
                flushBuffer(true);
                memoryContext.close();
                waitForPreviousUploadFinish();
            }
            catch (IOException | RuntimeException e) {
                abortUploadSuppressed(e);
                throw e;
            }

            try {
                uploadId.ifPresent(this::finishUpload);
            }
            catch (RuntimeException e) {
                abortUploadSuppressed(e);
                throw new IOException(e);
            }
        }

        private void ensureExtraBytesCapacity(int extraBytesCapacity)
        {
            int totalBytesCapacity = bufferSize + extraBytesCapacity;
            checkArgument(totalBytesCapacity <= partSize);
            if (buffer.length < totalBytesCapacity) {
                // buffer length might be 0
                int newBytesLength = max(buffer.length, initialBufferSize);
                if (totalBytesCapacity > newBytesLength) {
                    // grow array by 50%
                    newBytesLength = max(newBytesLength + (newBytesLength >> 1), totalBytesCapacity);
                    newBytesLength = min(newBytesLength, partSize);
                }
                buffer = Arrays.copyOf(buffer, newBytesLength);
                memoryContext.setBytes(buffer.length);
            }
        }

        private void flushBuffer(boolean finished)
                throws IOException
        {
            // Skip multipart upload if there would only be one part
            if (finished && !multipartUploadStarted) {
                InputStream in = new ByteArrayInputStream(buffer, 0, bufferSize);

                PutObjectRequest request = new PutObjectRequest(bucketName, key, in);
                ObjectMetadata metadata = new ObjectMetadata();
                metadata.setContentLength((long) bufferSize);
                metadata.setContentMd5(getMd5AsBase64(buffer, 0, bufferSize));
                metadata.setObjectStorageClass(obsStorageClass.getObsStorageClass());
                request.setMetadata(metadata);
                request.setAcl(obsAclType.getCannedACL());

                setSseHeader(request,this.sseEnabled,this.sseType,this.sseKmsKeyId,this.sseCKeyBase64);

                try {
                    obsClient.putObject(request);
                    return;
                }
                catch (ObsException e) {
                    failed = true;
                    throw new IOException(e);
                }
            }

            // The multipart upload API only accept the last part to be less than 5MB
            if (bufferSize == partSize || (finished && bufferSize > 0)) {
                byte[] data = buffer;
                int length = bufferSize;

                if (finished) {
                    this.buffer = null;
                }
                else {
                    this.buffer = new byte[0];
                    this.initialBufferSize = partSize;
                    bufferSize = 0;
                }
                memoryContext.setBytes(0);

                try {
                    waitForPreviousUploadFinish();
                }
                catch (IOException e) {
                    failed = true;
                    abortUploadSuppressed(e);
                    throw e;
                }
                multipartUploadStarted = true;
                inProgressUploadFuture = uploadExecutor.submit(() -> uploadPage(data, length));
            }
        }

        private void waitForPreviousUploadFinish()
                throws IOException
        {
            if (inProgressUploadFuture == null) {
                return;
            }

            try {
                inProgressUploadFuture.get();
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new InterruptedIOException();
            }
            catch (ExecutionException e) {
                throw new IOException("Streaming upload failed", e);
            }
        }

        private UploadPartResult uploadPage(byte[] data, int length)
        {
            if (!uploadId.isPresent()) {
                uploadId = Optional.of(uploadIdFactory.get());
            }

            currentPartNumber++;
            UploadPartRequest uploadRequest = new UploadPartRequest(bucketName, key);
            uploadRequest.setUploadId(uploadId.get());
            uploadRequest.setPartNumber(currentPartNumber);
            uploadRequest.setPartSize((long) length);
            uploadRequest.setInput(new ByteArrayInputStream(data, 0, length));
            uploadRequest.setContentMd5(getMd5AsBase64(data, 0, length));

            UploadPartResult partResult = obsClient.uploadPart(uploadRequest);
            parts.add(partResult);
            return partResult;
        }

        private void finishUpload(String uploadId)
        {
            List<PartEtag> etags = parts.stream()
                    .map(uploadPartResult -> new PartEtag(uploadPartResult.getEtag(), uploadPartResult.getPartNumber()))
                    .collect(toList());

            obsClient.completeMultipartUpload(new CompleteMultipartUploadRequest(bucketName, key, uploadId, etags));
            STATS.uploadSuccessful();
        }

        private void abortUpload()
        {
            STATS.uploadFailed();
            uploadId.ifPresent(id -> obsClient.abortMultipartUpload(new AbortMultipartUploadRequest(bucketName, key, id)));
        }

        @SuppressWarnings("ObjectEquality")
        private void abortUploadSuppressed(Throwable throwable)
        {
            try {
                abortUpload();
            }
            catch (Throwable t) {
                if (throwable != t) {
                    throwable.addSuppressed(t);
                }
            }
        }

    }


    private static class PrestoObsOutputStream
            extends FilterOutputStream
    {
        private final ObsClient obsClient;
        private final String bucketName;
        private final String key;
        private final File tempFile;
        private final boolean sseEnabled;
        private final PrestoObsSseType sseType;
        private final String sseKmsKeyId;
        private final String sseCKeyBase64;
        private final PrestoObsAclType obsAclType;
        private final PrestoObsStorageClass obsStorageClass;

        private boolean closed;

        public PrestoObsOutputStream(
                ObsClient obsClient,
                String bucketName,
                String key,
                File tempFile,
                boolean sseEnabled,
                PrestoObsSseType sseType,
                String sseKmsKeyId,
                String sseCKeyBase64,
                PrestoObsAclType obsAclType,
                PrestoObsStorageClass obsStorageClass)
                throws IOException
        {
            super(new BufferedOutputStream(Files.newOutputStream(requireNonNull(tempFile, "tempFile is null").toPath())));

            this.obsClient = requireNonNull(obsClient, "obsClient is null");
            this.bucketName = requireNonNull(bucketName, "bucketName is null");
            this.key = requireNonNull(key, "key is null");
            this.tempFile = tempFile;
            this.sseEnabled = sseEnabled;
            this.sseType = requireNonNull(sseType, "sseType is null");
            this.sseKmsKeyId = sseKmsKeyId;
            this.sseCKeyBase64 = sseCKeyBase64;
            this.obsAclType = obsAclType;
            this.obsStorageClass = obsStorageClass;

            log.debug("OutputStream for key '%s' using file: %s", key, tempFile);
        }

        @Override
        public void close()
                throws IOException
        {
            if (closed) {
                return;
            }
            closed = true;

            try {
                super.close();
                uploadObject();
            }
            finally {
                if (!tempFile.delete()) {
                    log.warn("Could not delete temporary file: %s", tempFile);
                }
            }
        }

        private void uploadObject()
                throws IOException
        {
            try {
                log.debug("Starting upload for host: %s, key: %s, file: %s, size: %s", bucketName, key, tempFile, tempFile.length());

                STATS.uploadStarted();
                PutObjectRequest request = new PutObjectRequest(bucketName, key, tempFile);
                ObjectMetadata metadata = new ObjectMetadata();
                metadata.setObjectStorageClass(obsStorageClass.getObsStorageClass());
                request.setMetadata(metadata);
                request.setAcl(obsAclType.getCannedACL());
                setSseHeader(request,this.sseEnabled,this.sseType,this.sseKmsKeyId,this.sseCKeyBase64);

                if (log.isDebugEnabled()) {
                    request.setProgressListener(progressStatus -> {
                        // 上传平均速率
                        log.debug("Upload averageSpeed  (%s/%s): %s" ,bucketName, key, progressStatus.getAverageSpeed());
                        // 上传进度百分比
                        log.debug("Upload percentage (%s/%s): %s" ,bucketName, key, progressStatus.getTransferPercentage());
                    });
                    request.setProgressInterval(1024 * 1024L);
                }
                obsClient.putObject(request);
                STATS.uploadSuccessful();
                log.debug("Completed upload for host: %s, key: %s", bucketName, key);
            }
            catch (ObsException e) {
                STATS.uploadFailed();
                throw new IOException(e);
            }
            catch (ServiceException e) {
                STATS.uploadFailed();
                throw new IOException(e);
            }
            catch (Exception e) {
                STATS.uploadFailed();
                Thread.currentThread().interrupt();
                throw new IOException(e);
            }
        }
    }

    @VisibleForTesting
    ObsClient getObsClient()
    {
        return obsClient;
    }

    @VisibleForTesting
    void setObsClient(ObsClient obsClient)
    {
        this.obsClient = obsClient;
    }

    public static String getBucketName(URI uri)
    {
        if (uri.getHost() != null) {
            return uri.getHost();
        }

        if (uri.getUserInfo() == null) {
            return uri.getAuthority();
        }

        throw new IllegalArgumentException("Unable to determine obs bucket from URI.");
    }

    public static PrestoObsFileSystemStats getFileSystemStats()
    {
        return STATS;
    }

    private InitiateMultipartUploadResult initMultipartUpload(String bucket, String key)
    {
        try {
            return retry()
                    .maxAttempts(maxAttempts)
                    .exponentialBackoff(BACKOFF_MIN_SLEEP, maxBackoffTime, maxRetryTime, 2.0)
                    .stopOn(InterruptedException.class, UnrecoverableObsOperationException.class, FileNotFoundException.class)
                    .onRetry(STATS::newInitiateMultipartUploadRetry)
                    .run("initiateMultipartUpload", () -> {
                        try {
                            InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucket, key);
                            ObjectMetadata metadata = new ObjectMetadata();
                            metadata.setObjectStorageClass(obsStorageClass.getObsStorageClass());
                            request.setMetadata(metadata);
                            request.setAcl(obsAclType.getCannedACL());
                            setSseHeader(request,this.sseEnabled,this.sseType,this.sseKmsKeyId,this.sseCKeyBase64);

                            return obsClient.initiateMultipartUpload(request);
                        }
                        catch (RuntimeException e) {
                            STATS.newInitiateMultipartUploadError();
                            if (e instanceof ObsException) {
                                ObsException obsException = (ObsException) e;
                                switch (obsException.getResponseCode()) {
                                    case HTTP_FORBIDDEN:
                                    case HTTP_BAD_REQUEST:
                                        throw new UnrecoverableObsOperationException(bucket, key, e);
                                    case HTTP_NOT_FOUND:
                                        throwIfFileNotFound(bucket, key, obsException);
                                        throw new UnrecoverableObsOperationException(bucket, key, e);
                                }
                            }
                            if (e instanceof ServiceException) {
                                ServiceException serviceException = (ServiceException) e;
                                switch (serviceException.getResponseCode()) {
                                    case HTTP_FORBIDDEN:
                                    case HTTP_BAD_REQUEST:
                                        throw new UnrecoverableObsOperationException(bucket, key, e);
                                    case HTTP_NOT_FOUND:
                                        throwIfFileNotFound(bucket, key, serviceException);
                                        throw new UnrecoverableObsOperationException(bucket, key, e);
                                }
                            }
                            throw e;
                        }
                    });
        }
        catch (InterruptedException | ObsException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
        catch (Exception e) {
            throwIfUnchecked(e);
            throw new RuntimeException(e);
        }
    }

    private static void setSseHeader(PutObjectBasicRequest request, boolean sseEnabled, PrestoObsSseType sseType, String sseKmsKeyId, String sseCKeyBase64){
        if (sseEnabled) {
            switch (sseType) {
                case KMS:
                    SseKmsHeader sseKmsHeader = new SseKmsHeader();
                    if (sseKmsKeyId != null) {
                        sseKmsHeader.setKmsKeyId(sseKmsKeyId);
                    }
                    request.setSseKmsHeader(sseKmsHeader);
                    break;
                case C:
                    if (sseCKeyBase64 != null) {
                        SseCHeader sseCHeader = new SseCHeader();
                        sseCHeader.setSseCKeyBase64(sseCKeyBase64);
                        request.setSseCHeader(sseCHeader);
                    }
                    break;
            }
        }
    }

    private static String getMd5AsBase64(byte[] data, int offset, int length)
    {
        @SuppressWarnings("deprecation")
        byte[] md5 = md5().hashBytes(data, offset, length).asBytes();
        return Base64.getEncoder().encodeToString(md5);
    }

    private static void throwIfFileNotFound(String bucket, String key, ServiceException serviceException)
            throws FileNotFoundException
    {
        String errorCode = serviceException.getErrorCode();
        if (NO_SUCH_KEY_ERROR_CODE.equals(errorCode) || NO_SUCH_BUCKET_ERROR_CODE.equals(errorCode)) {
            FileNotFoundException fileNotFoundException = new FileNotFoundException(format("%s (Bucket: %s, Key: %s)", firstNonNull(serviceException.getMessage(), serviceException), bucket, key));
            fileNotFoundException.initCause(serviceException);
            throw fileNotFoundException;
        }
    }

    private static void throwIfFileNotFound(String bucket, String key, ObsException obsException)
            throws FileNotFoundException
    {
        String errorCode = obsException.getErrorCode();
        if (NO_SUCH_KEY_ERROR_CODE.equals(errorCode) || NO_SUCH_BUCKET_ERROR_CODE.equals(errorCode)) {
            FileNotFoundException fileNotFoundException = new FileNotFoundException(format("%s (Bucket: %s, Key: %s)", firstNonNull(obsException.getMessage(), obsException), bucket, key));
            fileNotFoundException.initCause(obsException);
            throw fileNotFoundException;
        }
    }
}
